<?php
/**
 * @Author: gan
 * @Description:
 * @File:  RabbitImpl
 * @Version: 1.0.0
 * @Date: 2021/12/29 9:39 上午
 */
declare(strict_types=1);

namespace RX\Component\MQ\Impl;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire;
use RX\Enums\MQEnum;
use RX\Exception\BizException;
use RX\Helpers\DateHelper;
use RX\Helpers\RandomHelper;
use RX\Helpers\StringHelper;

class RabbitImpl implements MQInterface
{
    private AMQPStreamConnection $MQConn;                 // RabbitMQ连接管道
    private AMQPChannel          $MQChan;                 // RabbitMQ操作通道
    private string               $defaultType = "topic";  // 默认类型

    /**
     * MQTool constructor.
     * @param array $config
     */
    public function __construct(array $config = [])
    {
        if (empty($config)) {
            $config = StringHelper::config('mq.rabbit');
        }
        $this->MQConn = new AMQPStreamConnection(
            $config['host'],
            $config['port'],
            $config['username'],
            $config['password'],
            $config['vhost']
        );
        if ($this->MQConn->isConnected()) {
            $this->MQChan = $this->MQConn->channel();
        }
    }

    /**
     * exchange定义
     * @param string $exchangeName 交换机名称
     * @param string $type 类型
     * @param bool $durable 是否持久化
     * @param bool $autoDelete 是否自动删除
     * @param array $arguments 参数
     */
    public function exchangeDeclare(
        string $exchangeName,
        string $type,
        bool $durable,
        bool $autoDelete,
        array $arguments = array()
    ) {
        $this->MQChan->exchange_declare(
            $exchangeName,
            $type,
            false,
            $durable,
            $autoDelete,
            false,
            false,
            $arguments
        );
    }

    /**
     * 队列定义
     * @param string $queueName 队列名称
     * @param bool $durable 是否持久化
     * @param bool $autoDelete 是否自动删除
     * @param array $arguments 参数
     */
    public function queueDeclare(
        string $queueName,
        bool $durable,
        bool $autoDelete,
        $arguments = array()
    ) {
        $this->MQChan->queue_declare(
            $queueName,
            false,
            $durable,
            false,
            $autoDelete,
            false,
            $this->AMQPTable($arguments)
        );
    }

    /**
     * 队列绑定
     * @param string $queueName 队列名称
     * @param string $exchangeName 交换机名称
     * @param string $routingKey 路由key
     * @param array $arguments 参数
     */
    public function queueBind(
        string $queueName,
        string $exchangeName,
        string $routingKey,
        array $arguments = array()
    ) {
        $this->MQChan->queue_bind(
            $queueName,
            $exchangeName,
            $routingKey,
            false,
            $arguments
        );
    }

    /**
     * 向exchange发送消息
     * @param AMQPMessage $msg 信息
     * @param string $exchangeName 交换机名称
     * @param string $routingKey 路由key
     */
    public function publish(AMQPMessage $msg, string $exchangeName, string $routingKey)
    {
        $this->MQChan->basic_publish(
            $msg,
            $exchangeName,
            $routingKey
        );
    }

    /**
     * 消费
     * 是同步堵塞的，注意断线问题
     * @param string $queueName 队列名称
     * @param string $tag tag
     * @param bool $noAck 是否noAck
     * @param callable $callback 回调函数
     * @param array $arguments 参数
     * @param int|null $prefetchSize
     * @param int $prefetchCount
     * @param bool $nonBlocking 是否堵塞
     * @param int $timeOut wait等待时间
     * @throws \ErrorException
     */
    public function consumer(
        string $queueName,
        string $tag,
        bool $noAck,
        callable $callback,
        array $arguments = array(),
        int $prefetchSize = null,
        int $prefetchCount = 1,
        bool $nonBlocking = false,
        int $timeOut = 0
    ) {
        $this->MQChan->basic_qos($prefetchSize, $prefetchCount, null);
        $this->MQChan->basic_consume(
            $queueName,
            $tag,
            false,
            $noAck,
            false,
            false,
            $callback,
            null,
            $arguments
        );
        while (count($this->MQChan->callbacks)) {
            $this->MQChan->wait(null, $nonBlocking, $timeOut);
        }
    }

    /**
     * @param string $message 消息内容
     * @param array $properties 属性
     * @return AMQPMessage
     */
    public function AMQPMessage(string $message, array $properties = []): AMQPMessage
    {
        return new AMQPMessage($message, $properties);
    }

    /**
     * @param array $header
     * @return Wire\AMQPTable
     */
    public function AMQPTable(array $header): Wire\AMQPTable
    {
        return new Wire\AMQPTable($header);
    }

    /**
     * 生产者
     * @param array $params 数据
     * @return bool
     * @throws BizException
     */
    public function send(array $params): bool
    {
        $message      = $params["message"] ?? "";
        $exchangeName = $params["exchangeName"] ?? "";
        $queueName    = $params["queueName"] ?? "";
        $messageId    = $params["messageId"] ?? RandomHelper::makeSnowFlake() . "-" . DateHelper::getMicroTime();
        $args         = $params["args"] ?? array();

        if (empty($message) || empty($exchangeName) || empty($queueName)) {
            throw new BizException(MQEnum::class, MQEnum::PARAMS_ERROR);
        }

        $isSendOk = false;
        //confirm ack callback function
        $this->MQChan->set_ack_handler(function () use (&$isSendOk) {
            $isSendOk = true;
        });
        //confirm nack callback function
        $this->MQChan->set_nack_handler(function () use (&$isSendOk) {
            $isSendOk = false;
        });

        $routingKey = $params["routingKey"] ?? "{$exchangeName}.{$queueName}";
        $this->ExchangeDeclare($exchangeName, $this->defaultType, true, false);
        $this->QueueDeclare($queueName, true, false, $args);
        $this->QueueBind($queueName, $exchangeName, $routingKey);
        $message = $this->AMQPMessage(json_encode($message), ["message_id" => $messageId]);
        // 设置confirm消息确认模式并发送消息
        $this->MQChan->confirm_select();
        $this->publish($message, $exchangeName, $routingKey);
        // 等待mq服务给出确认收到消息的回复
        $this->MQChan->wait_for_pending_acks();
        return $isSendOk;
    }

    /**
     * 监听
     * @param array $params
     * @param callable $callback
     * @throws \ErrorException|BizException
     */
    public function listen(array $params, callable $callback)
    {
        $exchangeName = $params["exchangeName"] ?? "";
        $queueName    = $params["queueName"] ?? "";
        $args         = $params["args"] ?? array();

        if (empty($exchangeName) || empty($queueName)) {
            throw new BizException(MQEnum::class, MQEnum::PARAMS_ERROR);
        }

        $tag        = "{$queueName}";
        $routingKey = "{$exchangeName}.{$queueName}";
        $this->ExchangeDeclare($exchangeName, $this->defaultType, true, false);
        $this->QueueDeclare($queueName, true, false, $args);
        $this->QueueBind($queueName, $exchangeName, $routingKey);
        $this->consumer($queueName, $tag, false, $callback);
    }

    public function setDefaultType(string $name)
    {
        $this->defaultType = $name;
        return $this;
    }


    public function init(array $params)
    {
        return true;
    }

    public function close()
    {
        $this->MQChan->close();
        $this->MQConn->close();
    }

    public function __destruct()
    {
        $this->close();
    }
}
